package com.eyas.im.server.nio;

import com.eyas.im.protocol.decoder.Decoder;
import com.eyas.im.protocol.encoder.Encoder;
import com.eyas.im.server.IMServer;
import com.eyas.im.server.nio.handler.AbstractNioServerSocketHandler;
import com.eyas.im.server.nio.handler.ChannelCloseException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.lang.reflect.InvocationTargetException;
import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.net.StandardSocketOptions;
import java.nio.channels.*;
import java.util.Iterator;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

/**
 * 基于java NIO的server实现，非线程安全，一个通讯端只能启动一个
 */
public final class NIOIMServer implements IMServer {
    private final static Logger logger = LoggerFactory.getLogger(NIOIMServer.class);
    private NioServerConfiguration configuration;
    private ServerSocketChannel serverSocketChannel;
    private Selector selector;
    private volatile SERVERSTATUS loopFlag = SERVERSTATUS.READY; //execute无限循环的断开开关
    private int nioBugCounter = 0; //连续触发nioBug可能性判定的次数
    private long lastAcceptMils = 0L; //最后一次accept发生的次数
    private IOReactorThread[] ioReactorThreadPools;
    private ConcurrentLinkedQueue<SocketChannel> reactorIoEvents = new ConcurrentLinkedQueue<>();
    private int lastActiveReactorIndex = 0;//上一次被激活的IOReactor线程的索引
    private CountDownLatch countDownLatch; //用于在执行shutdown的时候，保证所有的资源都能够安全释放

    public NIOIMServer(String hostname, int port) {
        this.configuration = new NioServerConfiguration(hostname, port);
        validateConfiguration();
    }

    public NIOIMServer(NioServerConfiguration configuration) {
        this.configuration = configuration;
        validateConfiguration();
    }

    /**
     * 检查配置信息的正确性
     */
    private void validateConfiguration() {
        //1、如果NIO BUG检测启动，selector的block时长要大于Nio bug时间间隔阈值
        if (this.configuration.isNioBugChecker()
                && this.configuration.getSelectBlockTime() <= this.configuration.getNioBugIntervalThreshold())
            throw new IllegalArgumentException("SelectBlockTime要大于NioBugIntervalThreshold");

        //2、ioreactor thread的线程数量不要太多，暂定100个之内
        if (this.configuration.getEventThreadSize() > 100)
            throw new IllegalArgumentException("EventThreadSize的数量不要超过100个");

        //3、shutdown graceful中的等待时长至少要是selector block时间的3倍
        if (this.configuration.getShutdownWaitTime() / this.configuration.getSelectBlockTime() < 3)
            throw new IllegalArgumentException("ShutdownWaitTime的值至少是SelectBlockTime的3倍");

        //4、IOReactor线程从queue中读取待操作任务的阻塞时间，必须大于selector block时间，且不能超过shutdwon graceful的等待时长
        if (this.configuration.getIoReactorBlockTime() < this.configuration.getSelectBlockTime()
                || this.configuration.getIoReactorBlockTime() > this.configuration.getShutdownWaitTime())
            throw new IllegalArgumentException("IoReactorBlockTime的值要介于SelectBlockTime和ShutdownWaitTime之间");
    }

    @Override
    public void startUp() throws IOException {
        if (this.loopFlag != SERVERSTATUS.READY)
            throw new IllegalStateException("IMServer当前处于[" + this.loopFlag.name() + "]状态，无法启动");
        try {
            this.serverSocketChannel = ServerSocketChannel.open();
            serverSocketChannel.bind(new InetSocketAddress(this.configuration.getHostname(), this.configuration.getPort()));
            serverSocketChannel.configureBlocking(false);
            this.selector = Selector.open();
        } catch (IOException e) {
            if (this.selector != null)
                this.selector.close();
            if (this.serverSocketChannel != null)
                this.serverSocketChannel.close();
            throw e;
        }
        this.serverSocketChannel.register(this.selector, SelectionKey.OP_ACCEPT);
        this.loopFlag = SERVERSTATUS.START;
        this.execute();
    }

    private void execute() {
        this.ioReactorThreadPools = new IOReactorThread[this.configuration.getEventThreadSize()];
        int i = 0;
        for (; ; ) {
            if (i++ < this.configuration.getEventThreadSize()) {
                IOReactorThread ioReactorThread = new IOReactorThread("IMServer-reactor-" + i, Thread.currentThread().getThreadGroup());
                ioReactorThread.start();
                ioReactorThreadPools[i - 1] = ioReactorThread;
            } else
                break;
        }
        countDownLatch = new CountDownLatch(this.configuration.getEventThreadSize());
        try {
            while (this.loopFlag == SERVERSTATUS.START) {
                this.lastAcceptMils = System.currentTimeMillis();
                int acceptCount = this.selector.select(this.configuration.getSelectBlockTime());
                if (this.loopFlag == SERVERSTATUS.CLOSED)
                    break;
                if (acceptCount > 0) {
                    processAcceptSocket();
                } else {
                    //进行bug判定
                    if (this.loopFlag == SERVERSTATUS.START && IsjavaNioBugOccur()) {
                        //Nio bug发生，更改selector
                        this.selector = NIOIMServer.changeSelector(this.selector);
                    }
                }
            }
        } catch (IOException | SelectorChangeException e) {
            logger.error("IMServer执行过程中发生异常", e);
            this.shudownGraceful();
        }

        logger.debug("execute over......");
    }

    /**
     * accept到client的连接请求，处理请求
     */
    private void processAcceptSocket() throws IOException {
        Iterator<SelectionKey> iterator = this.selector.selectedKeys().iterator();
        while (iterator.hasNext()) {
            SelectionKey selectionKey = iterator.next();
            try {
                switch (selectionKey.readyOps()) {
                    case SelectionKey.OP_ACCEPT:
                        //获取到client的请求，修改socketChannel的配置，并注册到selector中
                        ServerSocketChannel serverSocketChannel = (ServerSocketChannel) selectionKey.channel();
                        for (; ; ) {
                            SocketChannel socketChannel = serverSocketChannel.accept();
                            if (socketChannel == null)
                                break;
                            socketChannel.configureBlocking(false);
                            socketChannel.setOption(StandardSocketOptions.SO_KEEPALIVE, true); //启动tcp keep-alive
                            reactorIoEvents.add(socketChannel);
                            this.wakeUpReactorThread();
                        }
                        break;
                    case SelectionKey.OP_CONNECT:
                        //这个事件不会在server端触发
                        logger.warn("imserver端的selector接收到OP_CONNECT事件");
                        break;
                    case SelectionKey.OP_READ:
                        //这个事件不会在server端触发
                        logger.warn("imserver端的selector接收到OP_READ事件");
                        break;
                    case SelectionKey.OP_WRITE:
                        //这个事件不会在server端触发
                        logger.warn("imserver端的selector接收到OP_WRITE事件");
                        break;
                }
            } finally {
                iterator.remove();
            }
        }
    }

    /**
     * 立即唤醒reactor线程来处理IO事件
     */
    private void wakeUpReactorThread() {
        int index = Math.abs(this.lastActiveReactorIndex++ % this.configuration.getEventThreadSize());
        this.ioReactorThreadPools[index].selector.wakeup();
    }

    /**
     * java NIO在linux epoll模型下，是否触发空轮询的bug
     *
     * @return
     */
    private boolean IsjavaNioBugOccur() {
        if (!this.configuration.isNioBugChecker())
            return false;
        if (System.currentTimeMillis() - this.lastAcceptMils <= this.configuration.getNioBugIntervalThreshold()) {
            this.nioBugCounter++;
        } else {
            this.nioBugCounter = 0;
        }
        return this.nioBugCounter >= this.configuration.getNioBugPossibleCount();
    }

    /**
     * 将现在的selector的所有注册信息，copy到新的selector中，并关闭旧的selector
     * 这个方法主要是当java nio bug出现时使用的
     */
    protected static Selector changeSelector(Selector oldSelector) throws SelectorChangeException {
        logger.warn("发生java NIO的空轮询bug，更换selector");
        try {
            Selector newSelector = Selector.open();
            for (SelectionKey selectionKey : oldSelector.keys()) {
                SelectableChannel channel = selectionKey.channel();
                int intrestOps = selectionKey.interestOps();
                Object attachment = selectionKey.attachment();
                selectionKey.cancel();
                channel.register(newSelector, intrestOps, attachment);
            }
            return newSelector;
        } catch (IOException e) {
            throw new SelectorChangeException("selector更换失败", e);
        } finally {
            try {
                oldSelector.close();
            } catch (IOException e) {
                logger.warn("select更换时，oldSelector关闭失败，这可能造成内存泄露问题", e);
            }
        }
    }

    @Override
    public void shudownGraceful() {
        this.loopFlag = SERVERSTATUS.CLOSED;
        this.selector.wakeup();
        for (int i = 0; i < this.ioReactorThreadPools.length; i++) {
            if (this.ioReactorThreadPools[i].isAlive()) {
                this.ioReactorThreadPools[i].selector.wakeup();//立即唤醒IOReactor线程阻塞中的selector
                this.ioReactorThreadPools[i].interrupt();
            }
        }
        try {
            countDownLatch.await(this.configuration.getShutdownWaitTime(), TimeUnit.MILLISECONDS);
            this.selector.close();
            this.serverSocketChannel.close();
            logger.debug("成功关闭IMServer");
        } catch (IOException | InterruptedException e) {
            e.printStackTrace();
        }
    }

    @Override
    public SocketAddress getSocketAddress() throws IOException {
        return this.serverSocketChannel.getLocalAddress();
    }

    private final class IOReactorThread extends Thread {
        Selector selector;
        private int likeNioBugCounter = 0;

        IOReactorThread(String threadName, ThreadGroup threadGroup) {
            super(threadGroup, threadName);
        }

        /**
         * 当前线程的Selector是否发生了空轮询bug
         *
         * @return
         */
        private boolean isNioLoopBugOccur(long lastProcessTime) {
            if (!NIOIMServer.this.configuration.isNioBugChecker())
                return false;
            if (System.currentTimeMillis() - lastProcessTime <= NIOIMServer.this.configuration.getNioBugIntervalThreshold()) {
                this.likeNioBugCounter++;
            } else {
                this.likeNioBugCounter = 0;
            }
            return this.likeNioBugCounter >= NIOIMServer.this.configuration.getNioBugPossibleCount();
        }

        @Override
        public void run() {
            try {
                this.selector = Selector.open();
                while (NIOIMServer.this.loopFlag == SERVERSTATUS.START) {
                    long lastProcessTime = System.currentTimeMillis(); //用于检测是否发生NIO空轮询bug
                    int selectSize = this.selector.select(NIOIMServer.this.configuration.getSelectBlockTime());
                    //server主动关闭
                    if (NIOIMServer.this.loopFlag == SERVERSTATUS.CLOSED) {
                        this.shutdownReactor();
                        break;
                    }
                    if (selectSize > 0) {
                        processIOEvent(this.selector);
                    } else {
                        //在这里判断，是否发生了空轮询bug
                        if (NIOIMServer.this.loopFlag == SERVERSTATUS.START && isNioLoopBugOccur(lastProcessTime)) {
                            //替换selector
                            this.selector = NIOIMServer.changeSelector(this.selector);
                        }
                    }

                    //提取accept的新的socketChannel
                    if (NIOIMServer.this.loopFlag == SERVERSTATUS.START)
                        processNewSocketChannel();
                }
            } catch (IOException e) {
                logger.error("IOReactor线程异常中断", e);
            } catch (SelectorChangeException e) {
                logger.error("IOReactor线程的Selector更换发生异常", e);
            } finally {
                NIOIMServer.this.countDownLatch.countDown();
                try {
                    this.selector.close();
                } catch (IOException e) {
                    logger.error("selector关闭异常", e);
                }
                logger.debug("{}执行结束", currentThread().getName());
            }
        }

        private void shutdownReactor() {
            for (SelectionKey selectionKey : this.selector.keys()) {
                SocketChannel socketChannel = (SocketChannel) selectionKey.channel();
                AbstractNioServerSocketHandler<?> adapter = (AbstractNioServerSocketHandler<?>) selectionKey.attachment();
                try {
                    adapter.beforeClose();
                    selectionKey.cancel();
                    socketChannel.close();
                } catch (IOException e) {
                    try {
                        logger.warn("server主动关闭时，关闭socket:[" + socketChannel.getRemoteAddress() + "]失败", e);
                    } catch (IOException ex) {
                        ex.printStackTrace();
                    }
                }
            }
            SocketChannel socketChannel = null;
            while ((socketChannel = NIOIMServer.this.reactorIoEvents.poll()) != null) {
                try {
                    socketChannel.close();
                } catch (IOException e) {
                    try {
                        logger.warn("server主动关闭时，关闭socket:[" + socketChannel.getRemoteAddress() + "]失败", e);
                    } catch (IOException ex) {
                        ex.printStackTrace();
                    }
                }
            }
        }

        /**
         * 读取全连接队列中的socketChannel,建立事件监听
         */
        private void processNewSocketChannel() {
            SocketChannel socketChannel = null;
            while ((socketChannel = NIOIMServer.this.reactorIoEvents.poll()) != null) {
                try {
                    AbstractNioServerSocketHandler<?> serverSocketHandler = NIOIMServer.this.configuration.getHandlerClass()
                            .getConstructor(Encoder.class, Decoder.class, int.class)
                            .newInstance(NIOIMServer.this.configuration.getEncoderClass().newInstance(),
                                    NIOIMServer.this.configuration.getDecoderClass().newInstance(),
                                    NIOIMServer.this.configuration.getBufferCapacity());
                    SelectionKey selectionKey = socketChannel.register(this.selector, SelectionKey.OP_READ);
                    selectionKey.attach(serverSocketHandler);
                    serverSocketHandler.bind(selectionKey, socketChannel);
                    serverSocketHandler.afterAccept(socketChannel.getRemoteAddress());
                } catch (InstantiationException | IllegalAccessException | NoSuchMethodException | InvocationTargetException e) {
                    logger.error("实例化ServerSocketHandler实例失败", e);
                    try {
                        //当前时刻,socketChannel还没有与selector关联
                        socketChannel.close();
                    } catch (IOException ex) {
                        ex.printStackTrace();
                    }
                } catch (IOException e) {
                    try {
                        logger.error("socketChannel[" + socketChannel.getRemoteAddress() + "]异常", e);
                        //当前时刻,socketChannel还没有与selector关联
                        socketChannel.close();
                    } catch (IOException ex) {
                        logger.error("socketChannel关闭失败", e);
                    }
                }
            }
        }

        /**
         * 处理具体的READ和WRITE事件
         *
         * @param selector
         */
        private void processIOEvent(Selector selector) {
            Iterator<SelectionKey> selectionKeyIterator = selector.selectedKeys().iterator();
            while (selectionKeyIterator.hasNext()) {
                try {
                    SelectionKey selectionKey = selectionKeyIterator.next();
                    SocketChannel socketChannel = (SocketChannel) selectionKey.channel();
                    switch (selectionKey.readyOps()) {
                        case SelectionKey.OP_ACCEPT:
                            logger.warn("IOReactor的selector获取到OP_ACCEPT");
                            break;
                        case SelectionKey.OP_CONNECT:
                            logger.warn("IOReactor的selector获取到OP_CONNECT");
                            break;
                        case SelectionKey.OP_READ:
                            logger.debug("IOReactor的selector获取到OP_READ");
                            AbstractNioServerSocketHandler<?> adapter = (AbstractNioServerSocketHandler<?>) selectionKey.attachment();
                            try {
                                adapter.onDataReceived(adapter.read());
                            } catch (IOException e) {
                                try {
                                    logger.warn("当前终端从[" + socketChannel.getRemoteAddress() + "]读消息时出现问题", e);
                                    adapter.beforeClose();
                                    selectionKey.cancel();
                                    socketChannel.close();
                                } catch (IOException ex) {
                                    logger.error("关闭终端失败", e);
                                }
                            } catch (ChannelCloseException e) {
                                logger.debug("client端主动断开");
                                adapter.beforeClose();
                                selectionKey.cancel();
                                try {
                                    socketChannel.close();
                                } catch (IOException ex) {
                                    logger.error("关闭终端失败", e);
                                }
                            }
                            break;
                        case SelectionKey.OP_WRITE:
                            logger.debug("IOReactor的selector获取到OP_WRITE");
                            AbstractNioServerSocketHandler<?> adapter_w = (AbstractNioServerSocketHandler<?>) selectionKey.attachment();
                            try {
                                adapter_w.write();
                            } catch (IOException e) {
                                try {
                                    logger.warn("当前终端向[" + socketChannel.getRemoteAddress() + "]写消息时出现问题", e);
                                    adapter_w.beforeClose();
                                    selectionKey.cancel();
                                    socketChannel.close();
                                } catch (IOException ex) {
                                    logger.error("关闭终端失败", e);
                                }
                            }
                            break;
                    }
                } finally {
                    selectionKeyIterator.remove();
                }
            }
        }
    }

}
